package com.ams.protocol.rtmp;

import java.io.DataInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;

import com.ams.io.BufferList;
import com.ams.io.BufferInputStream;
import com.ams.protocol.rtmp.amf.Amf0Deserializer;
import com.ams.protocol.rtmp.amf.Amf3Deserializer;
import com.ams.protocol.rtmp.amf.AmfException;
import com.ams.protocol.rtmp.amf.AmfSwitchToAmf3Exception;
import com.ams.protocol.rtmp.amf.AmfValue;
import com.ams.protocol.rtmp.message.RtmpMessage;
import com.ams.protocol.rtmp.message.RtmpMessageAbort;
import com.ams.protocol.rtmp.message.RtmpMessageAck;
import com.ams.protocol.rtmp.message.RtmpMessageAudio;
import com.ams.protocol.rtmp.message.RtmpMessageChunkSize;
import com.ams.protocol.rtmp.message.RtmpMessageCommand;
import com.ams.protocol.rtmp.message.RtmpMessageData;
import com.ams.protocol.rtmp.message.RtmpMessagePeerBandwidth;
import com.ams.protocol.rtmp.message.RtmpMessageSharedObject;
import com.ams.protocol.rtmp.message.RtmpMessageUnknown;
import com.ams.protocol.rtmp.message.RtmpMessageUserControl;
import com.ams.protocol.rtmp.message.RtmpMessageVideo;
import com.ams.protocol.rtmp.message.RtmpMessageWindowAckSize;
import com.ams.protocol.rtmp.so.SoMessage;

public class RtmpMessageDeserializer {
    private int readChunkSize = 128;
    protected HashMap<Integer, RtmpChunkDataReader> chunkDataMap;
    protected BufferInputStream in;

    public RtmpMessageDeserializer(BufferInputStream in) {
        super();
        this.in = in;
        this.chunkDataMap = new HashMap<Integer, RtmpChunkDataReader>();
    }

    public RtmpMessage read(RtmpHeader header) throws IOException,
            AmfException, RtmpException {
        int chunkStreamId = header.getChunkStreamId();
        RtmpChunkDataReader chunkData = chunkDataMap.get(chunkStreamId);
        if (chunkData == null) {
            chunkData = new RtmpChunkDataReader(header);
            chunkDataMap.put(chunkStreamId, chunkData);
        }
        int remain = chunkData.getChunkSize();
        if (remain > readChunkSize) {
            // continue to read
            chunkData.read(in, readChunkSize);
            return null;
        }
        chunkData.read(in, remain);
        // got a chunk data
        chunkDataMap.remove(chunkStreamId);
        return parseChunkData(chunkData);
    }

    private RtmpMessage parseChunkData(RtmpChunkDataReader chunk)
            throws IOException, AmfException, RtmpException {
        RtmpHeader header = chunk.getHeader();
        BufferList data = chunk.getChunkData();
        BufferInputStream bis = new BufferInputStream(data);
        switch (header.getType()) {
        case RtmpMessage.MESSAGE_USER_CONTROL:
            int event = bis.read16Bit();
            int streamId = -1;
            int timestamp = -1;
            switch (event) {
            case RtmpMessageUserControl.EVT_STREAM_BEGIN:
            case RtmpMessageUserControl.EVT_STREAM_EOF:
            case RtmpMessageUserControl.EVT_STREAM_DRY:
            case RtmpMessageUserControl.EVT_STREAM_IS_RECORDED:
                streamId = (int) bis.read32Bit();
                break;
            case RtmpMessageUserControl.EVT_SET_BUFFER_LENGTH:
                streamId = (int) bis.read32Bit();
                timestamp = (int) bis.read32Bit(); // buffer length
                break;
            case RtmpMessageUserControl.EVT_PING_REQUEST:
            case RtmpMessageUserControl.EVT_PING_RESPONSE:
                timestamp = (int) bis.read32Bit(); // timestamp
                break;
            default:
                event = RtmpMessageUserControl.EVT_UNKNOW;
            }
            return new RtmpMessageUserControl(event, streamId, timestamp);
        case RtmpMessage.MESSAGE_AMF3_COMMAND:
            bis.readByte(); // no used byte, continue to amf0 parsing
        case RtmpMessage.MESSAGE_AMF0_COMMAND: {
            DataInputStream dis = new DataInputStream(bis);
            Amf0Deserializer amf0Deserializer = new Amf0Deserializer(dis);
            Amf3Deserializer amf3Deserializer = new Amf3Deserializer(dis);

            String name = amf0Deserializer.read().string();
            int transactionId = amf0Deserializer.read().integer();
            ArrayList<AmfValue> argArray = new ArrayList<AmfValue>();
            boolean amf3Object = false;
            while (true) {
                try {
                    if (amf3Object) {
                        argArray.add(amf3Deserializer.read());
                    } else {
                        argArray.add(amf0Deserializer.read());
                    }
                } catch (AmfSwitchToAmf3Exception e) {
                    amf3Object = true;
                } catch (IOException e) {
                    break;
                }
            }

            AmfValue[] args = new AmfValue[argArray.size()];
            argArray.toArray(args);
            return new RtmpMessageCommand(name, transactionId, args);
        }

        case RtmpMessage.MESSAGE_VIDEO:
            return new RtmpMessageVideo(data);

        case RtmpMessage.MESSAGE_AUDIO:
            return new RtmpMessageAudio(data);

        case RtmpMessage.MESSAGE_AMF0_DATA:
        case RtmpMessage.MESSAGE_AMF3_DATA:
            return new RtmpMessageData(data);
        case RtmpMessage.MESSAGE_SHARED_OBJECT:
            SoMessage so = SoMessage.read(new DataInputStream(bis));
            return new RtmpMessageSharedObject(so);

        case RtmpMessage.MESSAGE_CHUNK_SIZE:
            readChunkSize = (int) bis.read32Bit();
            return new RtmpMessageChunkSize(readChunkSize);

        case RtmpMessage.MESSAGE_ABORT:
            return new RtmpMessageAbort((int) bis.read32Bit());

        case RtmpMessage.MESSAGE_ACK:
            return new RtmpMessageAck((int) bis.read32Bit());

        case RtmpMessage.MESSAGE_WINDOW_ACK_SIZE:
            return new RtmpMessageWindowAckSize((int) bis.read32Bit());

        case RtmpMessage.MESSAGE_PEER_BANDWIDTH:
            int windowAckSize = (int) bis.read32Bit();
            byte limitType = bis.readByte();
            return new RtmpMessagePeerBandwidth(windowAckSize, limitType);
            // case RtmpMessage.MESSAGE_AGGREGATE:

        default:
            return new RtmpMessageUnknown(header.getType(), data);
        }
    }

    public int getReadChunkSize() {
        return readChunkSize;
    }

    public void setReadChunkSize(int readChunkSize) {
        this.readChunkSize = readChunkSize;
    }
}
